package com.sunday.common.mq.rocket.brave.study.base.e10_Brave_Trace;

import brave.Tracing;
import lombok.extern.slf4j.Slf4j;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;


/**
 * {@link brave.spring.rabbit.TracingRabbitListenerAdvice#invoke(MethodInvocation)}
 */
@Slf4j
@Component
public class BraveTraceConsumer {

    @Bean
    public RocketTracing rocketTracing(Tracing tracing) {
        return new RocketTracing(tracing);
    }

    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer(RocketTracing rocketTracing) throws MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("trace_consumer_group_name");
        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.subscribe("TopicTest", "*");

        consumer.setConsumeMessageBatchMaxSize(10);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            msgs.stream().forEach(message -> rocketTracing.invoke(message, msg -> log.info("{}", msg)));
            // 标记该消息已经被成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();

        log.info("Consumer Started.");

        return consumer;
    }


}
